package vn.com.epi.re.data.training;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.Collection;
import java.util.List;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.google.common.base.Strings;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
import com.google.common.io.Closeables;
import com.google.common.primitives.Ints;

import vn.com.epi.re.ApplicationContext;
import vn.com.epi.re.data.Pair;
import vn.com.epi.re.table.ArticleTable;
import vn.com.epi.re.table.RecommendationTable;
import vn.com.epi.re.table.UserTable;
import vn.com.epi.re.tools.StringUtils;
import vn.com.epi.re.tools.TableUtils;

public class Trainer extends Configured implements Tool {

    private final Log log = LogFactory.getLog(Trainer.class);
    
    private static final long TIME_LIMIT = 108000; // 30 hours
    private static final int GROUP_SIZE = 1000;
    private static final int MAX_ARTICLES = 40000;
        
    private int nUser, nArticle;
    private int[][] similarity;
    private int[] articleIds, articleReading;
    private ArrayListMultimap<Integer, Pair> graph;
    private Map<String, Integer> uMap;
    private Map<Integer, Integer> aMap, catMap;    
    private ArrayListMultimap<String, Integer> userData;
    private ArrayListMultimap<Integer, Integer> articleByCategory;
    private Map<Integer, Integer> articleData;
    private Map<Integer, Long> firstVisit;
    
    /*
     * 
     * Chu y sua lai current time 
     * 
     */
    
    private long current = System.currentTimeMillis() / 1000L;
//    private long current = 1397952000;  
        
    public Trainer() {        
        uMap = Maps.newHashMap();
        aMap = Maps.newHashMap();
        setConf(ApplicationContext.get().getConf());
    }
    
    private void loadFirstVisit() {
        HTableInterface table = ApplicationContext.get().getTable(ArticleTable.NAME);
        firstVisit = Maps.newHashMap();
        try {
            Scan scan = new Scan();
            scan.addColumn(ArticleTable.FAM_MAIN, ArticleTable.COL_FIRST_VISIT_TIMESTAMPS);        
            ResultScanner rs = table.getScanner(scan);
            for (Result r:rs) {                
                int articleId = Bytes.toInt(r.getRow());
                long timestamp = Bytes.toLong(r.getValue(ArticleTable.FAM_MAIN, ArticleTable.COL_FIRST_VISIT_TIMESTAMPS));
                firstVisit.put(articleId, timestamp);
            }
        } catch (Exception e) {
            System.out.println("Error while load first visit timestamp: " + e.toString());
        } finally {
            TableUtils.closeQuietly(table);
        }                         
    }
    
    private void loadActionData() {        
        HTableInterface table = ApplicationContext.get().getTable(UserTable.NAME);
        userData = ArrayListMultimap.create();
        articleData = Maps.newHashMap();
        try {
            Scan scan = new Scan();
            scan.addColumn(UserTable.FAM_MAIN, UserTable.COL_VISITED_ARTICLES);        
            ResultScanner rs = table.getScanner(scan);
            try {
                for (Result r:rs) {
                    String userId = Bytes.toString(r.getRow());
                    String articleCol = Bytes.toString(r.getValue(UserTable.FAM_MAIN, UserTable.COL_VISITED_ARTICLES));
                    String[] articles = articleCol.split(",");
                    for (int i = 0; i < articles.length; i++) {
                        int articleId = Integer.parseInt(articles[i]);
                        long born = firstVisit.containsKey(articleId) ? firstVisit.get(articleId) : 0; 
                        if (current - born < TIME_LIMIT) {
                            userData.put(userId, articleId);
                            if (articleData.containsKey(articleId)) 
                                articleData.put(articleId, articleData.get(articleId) + 1);
                            else
                                articleData.put(articleId, 1);
                        }
                    }
                }
            } finally {
                rs.close();
            }
        } catch (Exception e) {
            System.out.println("Error while load data from hbase: " + e.toString());
            StringWriter sw = new StringWriter();
            PrintWriter pw = new PrintWriter(sw);
            e.printStackTrace(pw);
            System.out.println(sw.toString());
        } finally {
            TableUtils.closeQuietly(table);
        }                
    }
    
    private void initial() throws IOException {
        articleIds = new int[nArticle + 1];
        for (int article : aMap.keySet())
            articleIds[aMap.get(article)] = article;
    }
    
    private int encodeUser(String userId) {
        int userEnCode;
        if (uMap.containsKey(userId))
            userEnCode = uMap.get(userId);
        else {
            nUser++;
            userEnCode = nUser;
            uMap.put(userId, userEnCode);
        }        
        return userEnCode;
    }
    
    private int encodeArticle(int articleId) {
        int articleEnCode;
        if (aMap.containsKey(articleId))
            articleEnCode = aMap.get(articleId);
        else {
            nArticle++;
            articleEnCode = nArticle;
            aMap.put(articleId, articleEnCode);
        }
        return articleEnCode;
    }
    
    private void precomputedSimilarity() throws IOException {
        log.info("-----------* Precomputed Item-item similarity *------------");
        try {
            int nLines = 0;
            articleReading = new int[MAX_ARTICLES];
            for (String userId : userData.keySet()) {
                encodeUser(userId);
                Collection<Integer> set = userData.get(userId);
                int numberOfReading = set.size();
                if (10 <= numberOfReading && numberOfReading <= 100) {                    
                    for (int articleId : set) {
                        int categoryId = catMap.get(articleId);
                        if (categoryId > 0) {                        
                            encodeArticle(articleId);
                            nLines++;
                        }
                    }
                }
            }
            System.out.println("Number of lines: " + nLines);            
            System.out.println("Number of users: " + nUser);            
            System.out.println("Number of articles: " + nArticle);            
            similarity = new int[nArticle + 1][nArticle + 1];
            for (String userId : userData.keySet()) {                
                Collection<Integer> set = userData.get(userId);
                int numberOfReading = set.size();
                if (10 <= numberOfReading && numberOfReading <= 100) {
                    Multimap<Integer, Integer> multiMap = ArrayListMultimap.create();
                    for (int articleId : set) {
                        int categoryId = catMap.get(articleId);
                        if (categoryId > 0) {
                            int articleEnCode = encodeArticle(articleId);
                            multiMap.put(categoryId, articleEnCode);
                            articleReading[articleEnCode]++;
                        }
                    }
                    for (int categoryId : multiMap.keySet()) {
                        Collection<Integer> articleSet = multiMap.get(categoryId);
                        int size = articleSet.size();
                        if (size > 1) {
                            int[] articles = Ints.toArray(articleSet);
                            for (int i = 0; i < size - 1; i++)
                                for (int j = i + 1; j < size; j++) {
                                    int firstArticle = articles[i];
                                    int secondArticle = articles[j];
                                    if (firstArticle != secondArticle) {
                                        similarity[firstArticle][secondArticle]++;                                            
                                        similarity[secondArticle][firstArticle]++;
                                    }
                                }
                        }
                    }
                }
            }
            log.info("completed precomputed Item-item similarity !!!");
        } catch (Exception e) {
            log.info("Error while precomputed Item-item similarity: " + e.getMessage());
            System.out.println("Error while precomputed Item-item similarity: " + e.toString());
            System.exit(-1);
        } finally {
            userData = null;        
            articleData = null;
            initial();
            uMap = null;
            aMap = null;
        }
    }
    
    private void loadCategory() {
        HTableInterface table = ApplicationContext.get().getTable(ArticleTable.NAME);
        catMap = Maps.newHashMap();
        try {
            Scan scan = new Scan();
            scan.addColumn(ArticleTable.FAM_MAIN, ArticleTable.COL_CATEGORY_ID);        
            ResultScanner rs = table.getScanner(scan);
            for (Result r:rs) {                
                int articleId = Bytes.toInt(r.getRow());
                int categoryId = Bytes.toInt(r.getValue(ArticleTable.FAM_MAIN, ArticleTable.COL_CATEGORY_ID));
                catMap.put(articleId, categoryId);
            }
        } catch (Exception e) {
            System.out.println("Error while load category: " + e.toString());
        } finally {
            TableUtils.closeQuietly(table);
        }                 
    }
    
    private double discount(int numberRead) {
        double value = 0;
        if (numberRead <= 0)
            return 0;
        if (numberRead <= 10) 
            value = ((numberRead - 1) / 10 + 1) * 0.01 + (((numberRead - 1) % 10 + 1) * 0.001);
        else
            if (numberRead <= 20)
                value = ((numberRead - 1) / 10 + 1) * 0.1 + (((numberRead - 1) % 10 + 1) * 0.01);
            else
                value = 0.89 + numberRead * 0.0001;            
        if (value > 1)
            value = 1;
        return value;
    }
    
//    private double  tanimotoCoefficient(int i, int j) {
//        double score;
//        if (similarity[i][j] == 0)
//            score = 0;
//        else
//            score = (similarity[i][j] * 1.0) / ((articleReading[i] + articleReading[j] - similarity[i][j]) * 1.0);
//        score *= discount(articleReading[i]) * discount(articleReading[j]);
//        return score;
//        
//    }

    private double  euclideanDistance(int i, int j) {
        double score;
        if (similarity[i][j] == 0 || articleReading[i] == 0 || articleReading[j] == 0)
            score = 0;
        else
            score = (similarity[i][j] * 1.0) / (Math.sqrt(articleReading[i] * 1.0) * Math.sqrt(articleReading[j] * 1.0));
        score *= discount(articleReading[i]) * discount(articleReading[j]);
        return score;
        
    }

    
    private void createRelatedGraph() {
        log.info("-----------* Start create related graph *------------");
        articleByCategory = ArrayListMultimap.create();        
        graph = ArrayListMultimap.create();
        for (int i = 1; i <= nArticle; i++) {
            int category = catMap.get(articleIds[i]);
            if (category > 0)
                articleByCategory.put(category, i);
        }
        for (int categoryId: articleByCategory.keySet()) {
            Collection<Integer> articleSet = articleByCategory.get(categoryId);
            int size = articleSet.size();
            if (size > 1) {
                for (int i: articleSet) {                    
                    for (int j: articleSet) 
                        if (i != j) {
                            int u = articleIds[i];
                            int v = articleIds[j];
                            double score = euclideanDistance(i, j);
                            if (score > 0) {      
                                graph.put(u, new Pair(v, score));
                            }
                        }
                }
            }
        }
        log.info("-----------* Complete create related graph *------------");
    }
    
    private void importRecommendations() {
        HTableInterface table = ApplicationContext.get().getTable(RecommendationTable.NAME);
        List<Get> gets = Lists.newArrayList();
        List<Put> puts = Lists.newArrayList();
        log.info("start import new recommendations");
        try {
            int numberOfArticles = graph.keySet().size(); 
            for (int articleId: graph.keySet()) {
                numberOfArticles--;                
                byte[] row = Bytes.toBytes(articleId);
                Get get = new Get(row);
                gets.add(get);
                if ((gets.size() == GROUP_SIZE) || (numberOfArticles == 0)) {
                    Result[] results = table.get(gets);
                    for (int i = 0; i < results.length; i++)  {
                        byte[] rowKey = results[i].getRow() != null ? results[i].getRow() : gets.get(i).getRow();
                        if (rowKey != null) {
                            int article = Bytes.toInt(rowKey);
                            List<Pair> set = graph.get(article);
                            String re = Bytes.toString(results[i].getValue(RecommendationTable.FAM_MAIN, RecommendationTable.COL_RECOMMENDATIONS));
                            String ne = StringUtils.getSubSet(set, 5);
                            String neighbors = StringUtils.mergeRecommendations(re, ne, 5);
                            if (!Strings.isNullOrEmpty(neighbors.trim())) {
                                byte[] value = Bytes.toBytes(neighbors);
                                Put put = new Put(rowKey);
                                put.add(RecommendationTable.FAM_MAIN, RecommendationTable.COL_RECOMMENDATIONS, value);
                                puts.add(put);
                            }
                         }
                    }
                    gets = Lists.newArrayList();
                    if (puts.size() > 0) {
                        table.put(puts);
                        puts = Lists.newArrayList();
                    }
                }
            }
            log.info("completed import new recommendations");
        } catch (Exception e) {
            log.info("Error while import new recommendations");            
            System.out.println("Error while import new recommendations: " + e.toString());
        } finally {
            TableUtils.closeQuietly(table);
        }        
    }
    
    public void training() throws Exception {
        try {
            loadFirstVisit();
            loadActionData();            
            loadCategory();
            precomputedSimilarity();
            createRelatedGraph();
            importRecommendations();
        }  catch (Exception e) {
            System.out.println("Error while training data: " + e.toString());
        } 
    }
    
    private void writeCurrentTime(String path) {
        BufferedWriter writer = null;
        try {
            File file = new File(path);
            writer = new BufferedWriter(new FileWriter(file));
            long now = System.currentTimeMillis() / 1000L;
            writer.write(now + "\n");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (writer != null)
                    writer.close();
            } catch (Exception ex) {
            }
        }
    }
    
    @Override
    public int run(String[] args) throws Exception {
        if (args.length > 0)
            writeCurrentTime(args[0]);
        training();
        return 0;
    }

    public static void main(String[] args) throws Exception {
        long startTime = System.currentTimeMillis();
        ApplicationContext c = ApplicationContext.init();
        try {
            ToolRunner.run(new Trainer(), args);
        } finally {
            Closeables.closeQuietly(c);            
        }        
        long endTime = System.currentTimeMillis();
        System.out.println("total time used for training: " + (endTime - startTime) + " ms.");
    }
}
